package com.fannuo.demo.Util.hadoop;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapreduce.Job;

import java.io.*;

public class HadoopClient {

    private FileSystem fileSystem;

    private HadoopConfig config;
    public HadoopClient() {
        config = new HadoopConfig();
        init();
    }

    public HadoopClient(String host,int port) {
        config = new HadoopConfig(host,port);
        init();
    }
    private void init() {
        try {
            fileSystem = DistributedFileSystem.get(config.getConfiguration());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    public byte[] getFileToBytes(String filePath) throws IOException {
        FSDataInputStream inputStream = fileSystem.open(new Path(filePath));
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        IOUtils.copyBytes(inputStream,outputStream,1024);
        inputStream.close();
        outputStream.close();
        return outputStream.toByteArray();
    }
    public void putFile(String filePath,byte[] bytes) throws IOException {
        FSDataOutputStream outputStream = fileSystem.create(new Path(filePath),true);
        outputStream.write(bytes);
        outputStream.hflush();
        outputStream.hsync();
        outputStream.close();
    }
    public void putFile(String filePath, File file) throws IOException {
        FileInputStream inputStream = new FileInputStream(file);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        IOUtils.copyBytes(inputStream,outputStream,1024);
        putFile(filePath,outputStream.toByteArray());
    }
    public Job createJob(String name) {
        try {
        return Job.getInstance(config.getConfiguration(), name);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    public void close() {
        if(fileSystem != null) {
            try {
                fileSystem.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
